package cn.doitedu.data_snyc;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job03_PmsProduct2Hbase {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建 cdc 连接器表，映射业务库中的  realtimedw.pms_product 表
        tenv.executeSql(
                " create table pms_product_mysql(        "+
                        "     id bigint                         "+ // 商品id
                        "     ,brand_id bigint                  "+ // 品牌id
                        "     ,product_category_id bigint       "+ // 品类id
                        "     ,price decimal(10,2)              "+ // 价格
                        "     ,primary key(id) not enforced     "+
                        " ) WITH (                              "+
                        "     'connector' = 'mysql-cdc',        "+
                        "     'hostname' = 'doitedu',           "+
                        "     'port' = '3306',                  "+
                        "     'username' = 'root',              "+
                        "     'password' = 'root',              "+
                        "     'database-name' = 'realtimedw',   "+
                        "     'table-name' = 'pms_product'      "+
                        " )                                     "
        );


        // 建hbase连接器表，映射 hbase中的 dim_product_info
        tenv.executeSql(
                " create table dim_product_info_hbase(                             "+
                        "    product_id  bigint,                               "+
                        "    f row<brand_id bigint,product_category_id bigint,price decimal(10,2)>, "+
                        "    primary key(product_id) not enforced              "+
                        " ) WITH (                                             "+
                        "  'connector' = 'hbase-2.2',                          "+
                        "  'table-name' = 'dim_product_info',                  "+
                        "  'zookeeper.quorum' = 'doitedu:2181'                 "+
                        " )                                                    "
        );


        // insert  ...  select  ....
        tenv.executeSql("insert into dim_product_info_hbase " +
                "select " +
                "    id as product_id,  " +
                "    row(brand_id,product_category_id,price) as f " +
                "from pms_product_mysql");

    }

}
